package aeemqx

import (
	"context"
	"devops-super/internal/consts"
	"devops-super/internal/consts/cacheConsts"
	"devops-super/internal/dao"
	"devops-super/internal/model/do"
	"devops-super/internal/model/entity"
	"devops-super/internal/model/mymodel"
	"devops-super/internal/service"
	"devops-super/pb"
	"devops-super/utility/util/dbUtil"
	"devops-super/utility/util/minioUtil"
	"devops-super/utility/util/timeUtil"
	"devops-super/utility/util/waveFileSave"
	"encoding/hex"
	"errors"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/os/glog"
	"github.com/gogf/gf/v2/os/gtime"
	"github.com/gogf/gf/v2/util/gconv"
	"strings"
	"sync"
	"time"
)

var (
	LastExecutedClearTime map[string]time.Time
	ClearSeconds          = 600 // 每600秒才会清理一次
)

// 定义缓存结构，记录每个设备最后删除AST记录的时间
var (
	astCleanupCache = make(map[string]time.Time)
	astCleanupMutex sync.Mutex
)

func (s *sAeemqx) handleAeSensorDataV1(ctx context.Context, message *pb.SensorMessage) (err error) {
	if message == nil {
		err = errors.New("invalid message")
		return
	}
	senserData := message.GetAeSensorData()
	if senserData == nil {
		err = errors.New("invalid sensor data")
		return
	}

	isFinalWave := false
	waveObj := senserData.GetAeWaveData()
	if waveObj != nil {
		isFinalWave = waveObj.GetAeWaveEnd() == 1
	} else {
		isFinalWave = true
	}
	if !isFinalWave {
		err = errors.New("not final")
		return
	}

	sec := senserData.GetAeArrivingSec()
	checkSec := timeUtil.CheckAeSecValid(int(sec))
	if !checkSec {
		err = errors.New("invalid time " + gconv.String(sec))
		return err
	}
	secMicro := timeUtil.GetJoinSecNsToMicroSec(uint64(senserData.GetAeArrivingSec()), uint64(senserData.GetAeArrivingNs()))
	deviceNo := message.GetNodeId()
	deviceInfo, err := service.Device().GetByDeviceNo(ctx, deviceNo)
	if err != nil {
		return err
	}
	if deviceInfo.OpenStatus != consts.OpenStatusActive {
		err = errors.New("device is not open")
		return err
	}
	distributeTableObj, err := service.Device().GetDeviceCurrentDistributeTableInfo(ctx, deviceNo, int64(secMicro), consts.AeDistributeTableFlagAeParam, true)
	if err != nil || distributeTableObj == nil || distributeTableObj.Id == 0 {
		if err == nil {
			err = errors.New("invalid distributeTable")
		}
		return err
	}
	dbConnGroup := distributeTableObj.ServerKey
	db := g.DB(dbConnGroup)
	paramRow := do.IotAeParamDemo{
		DeviceNo:       deviceNo,
		AeMeasureSpeed: uint(senserData.AeMeasureSpeed),
		AeGainValue:    uint(senserData.AeGainValue),
		AeEnlargeValue: senserData.AeEnlargeValue,
		AeAmp:          senserData.AeAmp,
		AePower:        senserData.AePower,
		AeAsl:          senserData.AeAsl,
		AeRms:          senserData.AeRms,
		AeRingingNum:   uint(senserData.AeRingingNum),
		AeRisingNum:    uint(senserData.AeRisingNum),
		AeRisingTime:   uint(senserData.AeRisingTime),
		AeDurningTime:  uint(senserData.AeDurningTime),
		AePeakFreq:     uint(senserData.AePeakFreq),
		AeCentroidFre:  uint(senserData.AeCentroidFre),
		AeFre1BandPro:  uint(senserData.AeFre1BandPro),
		AeFre2BandPro:  uint(senserData.AeFre2BandPro),
		AeFre3BandPro:  uint(senserData.AeFre3BandPro),
		AeFre4BandPro:  uint(senserData.AeFre4BandPro),
		AeFre5BandPro:  uint(senserData.AeFre5BandPro),
		Sampletime:     secMicro,
		CreatedAt:      gtime.Now(),
		Status:         0,
	}
	_, err = db.Model(distributeTableObj.Table).Ctx(ctx).Insert(paramRow)
	if err != nil {
		if dbUtil.IsTableNotExistError(err) {
			err = service.Device().CreateDeviceDistributeDataTable(ctx, distributeTableObj)
			if err != nil {
				return err
			}
		} else {
			return err
		}
	}
	// 设置缓存
	formatType := service.Device().GetAeDataCacheFormatType(ctx, 0, 0)
	cacheKey := cacheConsts.GetDeviceDataKey(deviceNo, consts.AeDistributeTableFlagAeParam, formatType)
	err = service.GetCacheInstance().Set(ctx, cacheKey, paramRow, 0)
	return err
}

// implementReceiveParamDataAE 实现接收ae参数数据
func (s *sAeemqx) implementReceiveParamDataAE(ctx context.Context, message *pb.SensorMessage) (err error) {
	if message == nil {
		err = errors.New("invalid message")
		return
	}
	senserData := message.GetSensorData().GetAeSensorData()
	if senserData == nil {
		err = errors.New("invalid sensor data")
		return
	}
	sec := senserData.GetAeArrivingSec()
	checkSec := timeUtil.CheckAeSecValid(int(sec))
	if !checkSec {
		err = errors.New("invalid time " + gconv.String(sec))
		return err
	}
	secMicro := timeUtil.GetJoinSecNsToMicroSec(uint64(senserData.GetAeArrivingSec()), uint64(senserData.GetAeArrivingNs()))
	deviceNo := message.GetNodeId()
	deviceInfo, err := service.Device().GetByDeviceNo(ctx, deviceNo)
	if err != nil {
		return err
	}
	if deviceInfo.OpenStatus != consts.OpenStatusActive {
		err = errors.New("device is not open")
		return err
	}
	distributeTableObj, err := service.Device().GetDeviceCurrentDistributeTableInfo(ctx, deviceNo, int64(secMicro), consts.AeDistributeTableFlagAeParam, true)
	if err != nil || distributeTableObj == nil || distributeTableObj.Id == 0 {
		if err == nil {
			err = errors.New("invalid distributeTable")
		}
		return err
	}
	dbConnGroup := distributeTableObj.ServerKey
	db := g.DB(dbConnGroup)
	paramRow := do.IotAeParamDemo{
		DeviceNo:       deviceNo,
		AeMeasureSpeed: uint(senserData.AeMeasureSpeed),
		AeGainValue:    uint(senserData.AeGainValue),
		AeEnlargeValue: senserData.AeEnlargeValue,
		AeAmp:          senserData.AeAmp,
		AePower:        senserData.AePower,
		AeAsl:          senserData.AeAsl,
		AeRms:          senserData.AeRms,
		AeRingingNum:   uint(senserData.AeRingingNum),
		AeRisingNum:    uint(senserData.AeRisingNum),
		AeRisingTime:   uint(senserData.AeRisingTime),
		AeDurningTime:  uint(senserData.AeDurningTime),
		AePeakFreq:     uint(senserData.AePeakFreq),
		AeCentroidFre:  uint(senserData.AeCentroidFre),
		AeFre1BandPro:  uint(senserData.AeFre1BandPro),
		AeFre2BandPro:  uint(senserData.AeFre2BandPro),
		AeFre3BandPro:  uint(senserData.AeFre3BandPro),
		AeFre4BandPro:  uint(senserData.AeFre4BandPro),
		AeFre5BandPro:  uint(senserData.AeFre5BandPro),
		Sampletime:     secMicro,
		CreatedAt:      gtime.Now(),
		Status:         0,
	}
	_, err = db.Model(distributeTableObj.Table).Ctx(ctx).Insert(paramRow)
	if err != nil {
		if dbUtil.IsTableNotExistError(err) {
			err = service.Device().CreateDeviceDistributeDataTable(ctx, distributeTableObj)
			if err != nil {
				return err
			}
		} else {
			return err
		}
	}
	// 设置缓存
	formatType := service.Device().GetAeDataCacheFormatType(ctx, 0, 0)
	cacheKey := cacheConsts.GetDeviceDataKey(deviceNo, consts.AeDistributeTableFlagAeParam, formatType)
	err = service.GetCacheInstance().Set(ctx, cacheKey, paramRow, 0)
	return err
}

// implementReceiveParamDataVI 实现接收vi参数数据
func (s *sAeemqx) implementReceiveParamDataVI(ctx context.Context, message *pb.SensorMessage) (err error) {
	if message == nil {
		err = errors.New("invalid message")
		return
	}
	senserData := message.GetSensorData().GetViSensorParaData()
	if senserData == nil {
		err = errors.New("invalid sensor data")
		return
	}
	sec := senserData.GetViArrivingSec()
	checkSec := timeUtil.CheckAeSecValid(int(sec))
	if !checkSec {
		err = errors.New("invalid time " + gconv.String(sec))
		return err
	}
	secMicro := timeUtil.GetJoinSecNsToMicroSec(uint64(senserData.GetViArrivingSec()), uint64(senserData.GetViArrivingNs()))
	deviceNo := message.GetNodeId()
	deviceInfo, err := service.Device().GetByDeviceNo(ctx, deviceNo)
	if err != nil {
		return err
	}
	if deviceInfo.OpenStatus != consts.OpenStatusActive {
		err = errors.New("device is not open")
		return err
	}
	distributeTableObj, err := service.Device().GetDeviceCurrentDistributeTableInfo(ctx, deviceNo, int64(secMicro), consts.AeDistributeTableFlagViParam, true)
	if err != nil || distributeTableObj == nil || distributeTableObj.Id == 0 {
		err = errors.New("invalid distributeTable")
		return err
	}
	dbConnGroup := distributeTableObj.ServerKey
	db := g.DB(dbConnGroup)
	paramRow := do.IotViParamDemo{
		CreatedAt:          gtime.Now(),
		DeviceNo:           deviceNo,
		ViMeasureType:      senserData.ViMeasureType,
		ViMeasureFreq:      senserData.ViMeasureFreq,
		Sampletime:         secMicro,
		ViMaxAccX:          senserData.ViMaxAccX,
		ViMaxAccfreqX:      senserData.ViMaxAccfreqX,
		ViAccPeakX:         senserData.ViAccPeakX,
		ViAccAverageX:      senserData.ViAccAverageX,
		ViAccAbsAverageX:   senserData.ViAccAbsAverageX,
		ViAccRmsX:          senserData.ViAccRmsX,
		ViAccVarX:          senserData.ViAccVarX,
		ViAccStdX:          senserData.ViAccStdX,
		ViAccCrestX:        senserData.ViAccCrestX,
		ViAccPulseX:        senserData.ViAccPulseX,
		ViAccFormX:         senserData.ViAccFormX,
		ViAccAllowX:        senserData.ViAccAllowX,
		ViAccSkewX:         senserData.ViAccSkewX,
		ViAccKurtosisX:     senserData.ViAccKurtosisX,
		ViAccAverageFreqX:  senserData.ViAccAverageFreqX,
		ViAccCentroidFreqX: senserData.ViAccCentroidFreqX,
		ViAccRmsFreqX:      senserData.ViAccRmsFreqX,
		ViAccMsFreqX:       senserData.ViAccMsFreqX,
		ViAccVarFreqX:      senserData.ViAccVarFreqX,
		ViAccStdFreqX:      senserData.ViAccStdFreqX,
		ViVelocityMaxX:     senserData.ViVelocityMaxX,
		ViVelocityMaxfreqX: senserData.ViVelocityMaxfreqX,
		ViVelocityRmsX:     senserData.ViVelocityRmsX,
		ViMaxDistanceX:     senserData.ViMaxDistanceX,
		ViAngleX:           senserData.ViAngleX,
		ViMaxAccY:          senserData.ViMaxAccY,
		ViMaxAccfreqY:      senserData.ViMaxAccfreqY,
		ViAccPeakY:         senserData.ViAccPeakY,
		ViAccAverageY:      senserData.ViAccAverageY,
		ViAccAbsAverageY:   senserData.ViAccAbsAverageY,
		ViAccRmsY:          senserData.ViAccRmsY,
		ViAccVarY:          senserData.ViAccVarY,
		ViAccStdY:          senserData.ViAccStdY,
		ViAccCrestY:        senserData.ViAccCrestY,
		ViAccPulseY:        senserData.ViAccPulseY,
		ViAccFormY:         senserData.ViAccFormY,
		ViAccAllowY:        senserData.ViAccAllowY,
		ViAccSkewY:         senserData.ViAccSkewY,
		ViAccKurtosisY:     senserData.ViAccKurtosisY,
		ViAccAverageFreqY:  senserData.ViAccAverageFreqY,
		ViAccCentroidFreqY: senserData.ViAccCentroidFreqY,
		ViAccRmsFreqY:      senserData.ViAccRmsFreqY,
		ViAccMsFreqY:       senserData.ViAccMsFreqY,
		ViAccVarFreqY:      senserData.ViAccVarFreqY,
		ViAccStdFreqY:      senserData.ViAccStdFreqY,
		ViVelocityMaxY:     senserData.ViVelocityMaxY,
		ViVelocityMaxfreqY: senserData.ViVelocityMaxfreqY,
		ViVelocityRmsY:     senserData.ViVelocityRmsY,
		ViMaxDistanceY:     senserData.ViMaxDistanceY,
		ViAngleY:           senserData.ViAngleY,
		ViMaxAccZ:          senserData.ViMaxAccZ,
		ViMaxAccfreqZ:      senserData.ViMaxAccfreqZ,
		ViAccPeakZ:         senserData.ViAccPeakZ,
		ViAccAverageZ:      senserData.ViAccAverageZ,
		ViAccAbsAverageZ:   senserData.ViAccAbsAverageZ,
		ViAccRmsZ:          senserData.ViAccRmsZ,
		ViAccVarZ:          senserData.ViAccVarZ,
		ViAccStdZ:          senserData.ViAccStdZ,
		ViAccCrestZ:        senserData.ViAccCrestZ,
		ViAccPulseZ:        senserData.ViAccPulseZ,
		ViAccFormZ:         senserData.ViAccFormZ,
		ViAccAllowZ:        senserData.ViAccAllowZ,
		ViAccSkewZ:         senserData.ViAccSkewZ,
		ViAccKurtosisZ:     senserData.ViAccKurtosisZ,
		ViAccAverageFreqZ:  senserData.ViAccAverageFreqZ,
		ViAccCentroidFreqZ: senserData.ViAccCentroidFreqZ,
		ViAccRmsFreqZ:      senserData.ViAccRmsFreqZ,
		ViAccMsFreqZ:       senserData.ViAccMsFreqZ,
		ViAccVarFreqZ:      senserData.ViAccVarFreqZ,
		ViAccStdFreqZ:      senserData.ViAccStdFreqZ,
		ViVelocityMaxZ:     senserData.ViVelocityMaxZ,
		ViVelocityMaxfreqZ: senserData.ViVelocityMaxfreqZ,
		ViVelocityRmsZ:     senserData.ViVelocityRmsZ,
		ViMaxDistanceZ:     senserData.ViMaxDistanceZ,
		ViAngleZ:           senserData.ViAngleZ,
		ViTemperature:      senserData.ViTemperature,
		ViVoltage:          senserData.ViVoltage,
		Status:             0,
	}
	_, err = db.Model(distributeTableObj.Table).Ctx(ctx).Insert(paramRow)
	if err != nil {
		if dbUtil.IsTableNotExistError(err) {
			err = service.Device().CreateDeviceDistributeDataTable(ctx, distributeTableObj)
			if err != nil {
				return err
			}
		} else {
			return err
		}
	}
	// 设置缓存
	formatType := service.Device().GetAeDataCacheFormatType(ctx, 0, 0)
	cacheKey := cacheConsts.GetDeviceDataKey(deviceNo, consts.AeDistributeTableFlagViParam, formatType)
	_ = service.GetCacheInstance().Set(ctx, cacheKey, paramRow, 0)
	return nil
}

// implementReceiveParamDataAEVI 实现接收aevi参数数据
func (s *sAeemqx) implementReceiveParamDataAEVI(ctx context.Context, message *pb.SensorMessage) (err error) {
	if message == nil {
		err = errors.New("invalid message")
		return
	}
	senserDataVi := message.GetSensorData().GetViSensorParaData()
	if senserDataVi == nil {
		err = errors.New("invalid vi data")
		return
	}
	senserDataAe := message.GetSensorData().GetAeSensorData()
	if senserDataAe == nil {
		err = errors.New("invalid ae data")
		return
	}
	sec := message.GetSensorData().GetArrivingSec()
	ns := message.GetSensorData().GetArrivingNs()
	checkSec := timeUtil.CheckAeSecValid(int(sec))
	if !checkSec {
		err = errors.New("invalid time " + gconv.String(sec))
		return err
	}
	secMicro := timeUtil.GetJoinSecNsToMicroSec(uint64(sec), uint64(ns))
	deviceNo := message.GetNodeId()
	deviceInfo, err := service.Device().GetByDeviceNo(ctx, deviceNo)
	if err != nil {
		return err
	}
	if deviceInfo.OpenStatus != consts.OpenStatusActive {
		err = errors.New("device is not open")
		return err
	}
	distributeTableObj, err := service.Device().GetDeviceCurrentDistributeTableInfo(ctx, deviceNo, int64(secMicro), consts.AeDistributeTableFlagAeViParam, true)
	if err != nil || distributeTableObj == nil || distributeTableObj.Id == 0 {
		err = errors.New("invalid distributeTable")
		return err
	}
	dbConnGroup := distributeTableObj.ServerKey
	db := g.DB(dbConnGroup)
	paramRow := do.IotAeViParamDemo{
		CreatedAt:          gtime.Now(),
		DeviceNo:           deviceNo,
		ViMeasureType:      senserDataVi.ViMeasureType,
		ViMeasureFreq:      senserDataVi.ViMeasureFreq,
		Sampletime:         secMicro,
		ViMaxAccX:          senserDataVi.ViMaxAccX,
		ViMaxAccfreqX:      senserDataVi.ViMaxAccfreqX,
		ViAccPeakX:         senserDataVi.ViAccPeakX,
		ViAccAverageX:      senserDataVi.ViAccAverageX,
		ViAccAbsAverageX:   senserDataVi.ViAccAbsAverageX,
		ViAccRmsX:          senserDataVi.ViAccRmsX,
		ViAccVarX:          senserDataVi.ViAccVarX,
		ViAccStdX:          senserDataVi.ViAccStdX,
		ViAccCrestX:        senserDataVi.ViAccCrestX,
		ViAccPulseX:        senserDataVi.ViAccPulseX,
		ViAccFormX:         senserDataVi.ViAccFormX,
		ViAccAllowX:        senserDataVi.ViAccAllowX,
		ViAccSkewX:         senserDataVi.ViAccSkewX,
		ViAccKurtosisX:     senserDataVi.ViAccKurtosisX,
		ViAccAverageFreqX:  senserDataVi.ViAccAverageFreqX,
		ViAccCentroidFreqX: senserDataVi.ViAccCentroidFreqX,
		ViAccRmsFreqX:      senserDataVi.ViAccRmsFreqX,
		ViAccMsFreqX:       senserDataVi.ViAccMsFreqX,
		ViAccVarFreqX:      senserDataVi.ViAccVarFreqX,
		ViAccStdFreqX:      senserDataVi.ViAccStdFreqX,
		ViVelocityMaxX:     senserDataVi.ViVelocityMaxX,
		ViVelocityMaxfreqX: senserDataVi.ViVelocityMaxfreqX,
		ViVelocityRmsX:     senserDataVi.ViVelocityRmsX,
		ViMaxDistanceX:     senserDataVi.ViMaxDistanceX,
		ViAngleX:           senserDataVi.ViAngleX,
		ViMaxAccY:          senserDataVi.ViMaxAccY,
		ViMaxAccfreqY:      senserDataVi.ViMaxAccfreqY,
		ViAccPeakY:         senserDataVi.ViAccPeakY,
		ViAccAverageY:      senserDataVi.ViAccAverageY,
		ViAccAbsAverageY:   senserDataVi.ViAccAbsAverageY,
		ViAccRmsY:          senserDataVi.ViAccRmsY,
		ViAccVarY:          senserDataVi.ViAccVarY,
		ViAccStdY:          senserDataVi.ViAccStdY,
		ViAccCrestY:        senserDataVi.ViAccCrestY,
		ViAccPulseY:        senserDataVi.ViAccPulseY,
		ViAccFormY:         senserDataVi.ViAccFormY,
		ViAccAllowY:        senserDataVi.ViAccAllowY,
		ViAccSkewY:         senserDataVi.ViAccSkewY,
		ViAccKurtosisY:     senserDataVi.ViAccKurtosisY,
		ViAccAverageFreqY:  senserDataVi.ViAccAverageFreqY,
		ViAccCentroidFreqY: senserDataVi.ViAccCentroidFreqY,
		ViAccRmsFreqY:      senserDataVi.ViAccRmsFreqY,
		ViAccMsFreqY:       senserDataVi.ViAccMsFreqY,
		ViAccVarFreqY:      senserDataVi.ViAccVarFreqY,
		ViAccStdFreqY:      senserDataVi.ViAccStdFreqY,
		ViVelocityMaxY:     senserDataVi.ViVelocityMaxY,
		ViVelocityMaxfreqY: senserDataVi.ViVelocityMaxfreqY,
		ViVelocityRmsY:     senserDataVi.ViVelocityRmsY,
		ViMaxDistanceY:     senserDataVi.ViMaxDistanceY,
		ViAngleY:           senserDataVi.ViAngleY,
		ViMaxAccZ:          senserDataVi.ViMaxAccZ,
		ViMaxAccfreqZ:      senserDataVi.ViMaxAccfreqZ,
		ViAccPeakZ:         senserDataVi.ViAccPeakZ,
		ViAccAverageZ:      senserDataVi.ViAccAverageZ,
		ViAccAbsAverageZ:   senserDataVi.ViAccAbsAverageX,
		ViAccRmsZ:          senserDataVi.ViAccRmsZ,
		ViAccVarZ:          senserDataVi.ViAccVarZ,
		ViAccStdZ:          senserDataVi.ViAccStdZ,
		ViAccCrestZ:        senserDataVi.ViAccCrestZ,
		ViAccPulseZ:        senserDataVi.ViAccPulseZ,
		ViAccFormZ:         senserDataVi.ViAccFormZ,
		ViAccAllowZ:        senserDataVi.ViAccAllowZ,
		ViAccSkewZ:         senserDataVi.ViAccSkewZ,
		ViAccKurtosisZ:     senserDataVi.ViAccKurtosisZ,
		ViAccAverageFreqZ:  senserDataVi.ViAccAverageFreqZ,
		ViAccCentroidFreqZ: senserDataVi.ViAccCentroidFreqZ,
		ViAccRmsFreqZ:      senserDataVi.ViAccRmsFreqZ,
		ViAccMsFreqZ:       senserDataVi.ViAccMsFreqZ,
		ViAccVarFreqZ:      senserDataVi.ViAccVarFreqZ,
		ViAccStdFreqZ:      senserDataVi.ViAccStdFreqZ,
		ViVelocityMaxZ:     senserDataVi.ViVelocityMaxZ,
		ViVelocityMaxfreqZ: senserDataVi.ViVelocityMaxfreqZ,
		ViVelocityRmsZ:     senserDataVi.ViVelocityRmsZ,
		ViMaxDistanceZ:     senserDataVi.ViMaxDistanceZ,
		ViAngleZ:           senserDataVi.ViAngleZ,
		ViTemperature:      senserDataVi.ViTemperature,
		ViVoltage:          senserDataVi.ViVoltage,
		AeMeasureSpeed:     senserDataAe.AeMeasureSpeed,
		AeGainValue:        senserDataAe.AeGainValue,
		AeEnlargeValue:     senserDataAe.AeEnlargeValue,
		AeAmp:              senserDataAe.AeAmp,
		AePower:            senserDataAe.AePower,
		AeAsl:              senserDataAe.AeAsl,
		AeRms:              senserDataAe.AeRms,
		AeRingingNum:       senserDataAe.AeRingingNum,
		AeRisingNum:        senserDataAe.AeRisingNum,
		AeRisingTime:       senserDataAe.AeRisingTime,
		AeDurningTime:      senserDataAe.AeDurningTime,
		AePeakFreq:         senserDataAe.AePeakFreq,
		AeCentroidFre:      senserDataAe.AeCentroidFre,
		AeFre1BandPro:      senserDataAe.AeFre1BandPro,
		AeFre2BandPro:      senserDataAe.AeFre2BandPro,
		AeFre3BandPro:      senserDataAe.AeFre3BandPro,
		AeFre4BandPro:      senserDataAe.AeFre4BandPro,
		AeFre5BandPro:      senserDataAe.AeFre5BandPro,
		Status:             0,
	}
	_, err = db.Model(distributeTableObj.Table).Ctx(ctx).Insert(paramRow)
	if err != nil {
		if dbUtil.IsTableNotExistError(err) {
			err = service.Device().CreateDeviceDistributeDataTable(ctx, distributeTableObj)
			if err != nil {
				return err
			}
		} else {
			return err
		}
	}
	// 设置缓存
	formatType := service.Device().GetAeDataCacheFormatType(ctx, 0, 0)
	cacheKey := cacheConsts.GetDeviceDataKey(deviceNo, consts.AeDistributeTableFlagAeViParam, formatType)
	_ = service.GetCacheInstance().Set(ctx, cacheKey, paramRow, 0)
	return nil
}

// implementReceiveWaveDataAE 实现接收声发射波形数据
func (s *sAeemqx) implementReceiveWaveDataAE(ctx context.Context, message *pb.SensorMessage) (err error) {
	if message == nil {
		err = errors.New("invalid message")
		return
	}
	deviceNo := message.GetNodeId()
	waveData := message.GetAeWave()
	sec := waveData.GetArrivingSec()
	ns := waveData.GetArrivingNs()
	checkSec := timeUtil.CheckAeSecValid(int(sec))
	if !checkSec {
		return errors.New("invalid sec")
	}
	secMicro := timeUtil.GetJoinSecNsToMicroSec(uint64(sec), uint64(ns))
	if waveData.GetWaveData() == nil {
		return errors.New("invalid wave data")
	}
	deviceInfo, err := service.Device().GetByDeviceNo(ctx, deviceNo)
	if err != nil {
		return err
	}
	if deviceInfo.OpenStatus != consts.OpenStatusActive {
		err = errors.New("device is not open")
		return err
	}
	distributeTableObj, err := service.Device().GetDeviceCurrentDistributeTableInfo(ctx, deviceNo, int64(secMicro), consts.AeDistributeTableFlagAeWave, true)
	if err != nil || distributeTableObj == nil || distributeTableObj.Id == 0 {
		err = errors.New("invalid distributeTable")
		return err
	}
	dbConnGroup := distributeTableObj.ServerKey
	db := g.DB(dbConnGroup)
	isFinalWave := waveData.GetWaveEnd() == 1
	// 如果是lora分段传的,需要插入临时表
	if deviceInfo.NetworkType == consts.ProductJointNetworkTypeLora {
		if len(waveData.GetWaveData()) > 255 { //
			err = errors.New("lora wave length exceeds 255")
			return err
		}
		tmp := do.IotAeWaveDemoTemplate{
			DeviceNo:   deviceNo,
			Sampletime: secMicro,
			CreatedAt:  gtime.Now(),
			WaveType:   consts.WaveTypeTimeDomain,
			Num:        waveData.GetFrameNum(),
			WaveData:   waveData.GetWaveData(),
		}
		// 保存到临时波形表
		_, err = db.Model(consts.DistributeTableAeWaveTemplate).Ctx(ctx).Insert(tmp)
		if err != nil {
			return
		}
	}
	if !isFinalWave {
		return
	}
	var wholeWave []byte
	if deviceInfo.NetworkType == consts.ProductJointNetworkTypeLora {
		// 从临时表取出波形 组合一个 插入波形表
		var templateWaves []*entity.IotAeWaveDemoTemplate
		err := db.Model(consts.DistributeTableAeWaveTemplate).Ctx(ctx).
			Where(dao.IotAeWaveDemoTemplate.Columns().DeviceNo, deviceNo).
			Where(dao.IotAeWaveDemoTemplate.Columns().Sampletime, secMicro).
			Where(dao.IotAeWaveDemoTemplate.Columns().WaveType, consts.WaveTypeTimeDomain).
			OrderAsc("num").Scan(&templateWaves)
		if err != nil {
			return err
		}
		for _, item := range templateWaves {
			wholeWave = append(wholeWave, item.WaveData...)
		}
	} else {
		wholeWave = waveData.GetWaveData()
	}
	lenWholeWave := len(wholeWave)
	if lenWholeWave == 0 {
		return
	}
	bucket := consts.BucketDefaultAeWave
	saveFlag := waveFileSave.GetSaveFlag(ctx)
	ymdhm := timeUtil.ConvertTimeIntToYmdhmDir(int(sec))
	unitype := gconv.String(consts.WaveTypeTimeDomain)
	miniopath := gconv.String(deviceInfo.DeptId) + "/" + deviceNo + "/" + unitype + "/" + ymdhm + "/" + gconv.String(secMicro) + consts.WaveSuffixAe
	if saveFlag == consts.StoreFlagOss {
		minioconf := minioUtil.GetMinioConfigByServername(ctx, deviceInfo.OssServer)
		if minioconf == nil || minioconf.Bucket == "" {
			return errors.New("minio config err")
		}
		bucket = minioconf.Bucket
		err = s.SaveWaveToOss(ctx, deviceInfo.OssServer, bucket, miniopath, wholeWave)
		if err != nil {
			return err
		}
	} else {
		fullpath, err := waveFileSave.GetWaveSavePath(bucket, miniopath)
		if err != nil {
			return err
		}
		err = waveFileSave.WriteToFile(fullpath, wholeWave, true)
		if err != nil {
			return err
		}
	}

	newWaveRow := do.IotAeWaveDemo{
		DeviceNo:    deviceNo,
		Sampletime:  secMicro,
		CreatedAt:   gtime.Now(),
		WaveType:    consts.WaveTypeTimeDomain,
		Flag:        saveFlag,
		StoreServer: deviceInfo.OssServer,
		Bucket:      bucket,
		Path:        miniopath,
		Status:      0,
	}
	_, err = db.Model(distributeTableObj.Table).Ctx(ctx).Insert(newWaveRow)
	if err != nil {
		return err
	}

	// 设置缓存
	formatType := service.Device().GetAeDataCacheFormatType(ctx, consts.WaveTypeTimeDomain, 0)
	cacheKey := cacheConsts.GetDeviceDataKey(deviceNo, consts.AeDistributeTableFlagAeWave, formatType)
	cacheData := &mymodel.WaveCacheData{
		Wave:       hex.EncodeToString(wholeWave),
		Sampletime: secMicro,
	}
	_ = service.GetCacheInstance().Set(ctx, cacheKey, cacheData, 0)

	// 清理临时表，但是不要每次都清理，以免io太多
	if deviceInfo.NetworkType == consts.ProductJointNetworkTypeLora {
		clearMapKey := dbConnGroup + consts.DistributeTableAeWaveTemplate
		lastTime, ok := LastExecutedClearTime[clearMapKey]
		if !ok || time.Now().Unix()-lastTime.Unix() > int64(ClearSeconds) {
			LastExecutedClearTime[clearMapKey] = time.Now()
			past := time.Now().Add(-time.Duration(ClearSeconds) * time.Second)
			timeLayout := "2006-01-02 15:04:05"
			timeStr := past.Local().Format(timeLayout)
			_, err = db.Model(consts.DistributeTableAeWaveTemplate).Ctx(ctx).Where(dao.IotAeWaveDemoTemplate.Columns().CreatedAt+"<", timeStr).Delete()
			if err != nil {
				return err
			}
		}
	}

	return nil
}

// implementReceiveWaveDataVI 实现接收vi波形数据
func (s *sAeemqx) implementReceiveWaveDataVI(ctx context.Context, message *pb.SensorMessage) (err error) {
	if message == nil {
		err = errors.New("invalid message")
		return
	}
	deviceNo := message.GetNodeId()
	waveData := message.GetViSensorWaveData()
	sec := waveData.GetViArrivingSec()
	ns := waveData.GetViArrivingNs()
	checkSec := timeUtil.CheckAeSecValid(int(sec))
	if !checkSec {
		err = errors.New("invalid sec")
		return err
	}
	secMicro := timeUtil.GetJoinSecNsToMicroSec(uint64(sec), uint64(ns))
	if waveData.GetViWaveData() == nil {
		return errors.New("invalid vi wave")
	}
	deviceInfo, err := service.Device().GetByDeviceNo(ctx, deviceNo)
	if err != nil {
		return err
	}
	if deviceInfo.OpenStatus != consts.OpenStatusActive {
		err = errors.New("device is not open")
		return err
	}
	distributeTableObj, err := service.Device().GetDeviceCurrentDistributeTableInfo(ctx, deviceNo, int64(secMicro), consts.AeDistributeTableFlagViwave, true)
	if err != nil || distributeTableObj == nil || distributeTableObj.Id == 0 {
		err = errors.New("invalid distributeTable")
		return err
	}
	dbConnGroup := distributeTableObj.ServerKey
	db := g.DB(dbConnGroup)
	isFinalWave := waveData.GetViWaveEnd() == 1
	// 如果是lora分段传的,需要插入临时表
	if deviceInfo.NetworkType == consts.ProductJointNetworkTypeLora {
		if len(waveData.GetViWaveData()) > 255 { //
			err = errors.New("lora wave length exceeds 255")
			return err
		}
		tmp := do.IotViWaveDemoTemplate{
			DeviceNo:   deviceNo,
			Sampletime: secMicro,
			CreatedAt:  gtime.Now(),
			Direction:  waveData.GetViWaveDirection(),
			WaveType:   waveData.GetViWaveType(),
			Num:        waveData.GetViFrameNum(),
			WaveData:   waveData.GetViWaveData(),
		}
		// 保存到临时波形表
		_, err = db.Model(consts.DistributeTableViWaveTemplate).Ctx(ctx).Insert(tmp)
		if err != nil {
			return
		}
	}
	if !isFinalWave {
		return
	}
	var wholeWave []byte
	if deviceInfo.NetworkType == consts.ProductJointNetworkTypeLora {
		// 从临时表取出波形 组合一个 插入波形表
		var templateWaves []*entity.IotViWaveDemoTemplate
		err := db.Model(consts.DistributeTableViWaveTemplate).Ctx(ctx).
			Where(dao.IotViWaveDemoTemplate.Columns().DeviceNo, deviceNo).
			Where(dao.IotViWaveDemoTemplate.Columns().Sampletime, secMicro).
			Where(dao.IotViWaveDemoTemplate.Columns().WaveType, waveData.GetViWaveType()).
			Where(dao.IotViWaveDemoTemplate.Columns().Direction, waveData.GetViWaveDirection()).
			OrderAsc("num").Scan(&templateWaves)
		if err != nil {
			return err
		}
		for _, item := range templateWaves {
			wholeWave = append(wholeWave, item.WaveData...)
		}
	} else {
		wholeWave = waveData.GetViWaveData()
	}
	lenWholeWave := len(wholeWave)
	if lenWholeWave == 0 {
		return nil
	}
	bucket := consts.BucketDefaultViWave
	saveFlag := waveFileSave.GetSaveFlag(ctx)

	ymdhm := timeUtil.ConvertTimeIntToYmdhmDir(int(sec))
	unitype := gconv.String(waveData.GetViWaveType()) + "_" + gconv.String(waveData.GetViWaveDirection())
	miniopath := gconv.String(deviceInfo.DeptId) + "/" + deviceNo + "/" + unitype + "/" + ymdhm + "/" + gconv.String(secMicro) + consts.WaveSuffixVi
	if saveFlag == consts.StoreFlagOss {
		minioconf := minioUtil.GetMinioConfigByServername(ctx, deviceInfo.OssServer)
		if minioconf == nil || minioconf.Bucket == "" {
			return errors.New("minioconf empty")
		}
		bucket = minioconf.Bucket
		_ = s.SaveWaveToOss(ctx, deviceInfo.OssServer, bucket, miniopath, wholeWave)
	} else {
		fullpath, err := waveFileSave.GetWaveSavePath(bucket, miniopath)
		if err != nil {
			return err
		}
		err = waveFileSave.WriteToFile(fullpath, wholeWave, true)
		if err != nil {
			return err
		}
	}
	newWaveRow := do.IotViWaveDemo{
		DeviceNo:    deviceNo,
		Sampletime:  secMicro,
		CreatedAt:   gtime.Now(),
		Direction:   waveData.GetViWaveDirection(),
		WaveType:    waveData.GetViWaveType(),
		Flag:        saveFlag,
		StoreServer: deviceInfo.OssServer,
		Bucket:      bucket,
		Path:        miniopath,
		Status:      nil,
	}
	_, err = db.Model(distributeTableObj.Table).Ctx(ctx).Insert(newWaveRow)
	if err != nil {
		return err
	}

	// 设置缓存
	formatType := service.Device().GetAeDataCacheFormatType(ctx, int(waveData.GetViWaveType()), int(waveData.GetViWaveDirection()))
	cacheKey := cacheConsts.GetDeviceDataKey(deviceNo, consts.AeDistributeTableFlagViwave, formatType)
	cacheData := &mymodel.WaveCacheData{
		Wave:       hex.EncodeToString(wholeWave),
		Sampletime: secMicro,
	}
	_ = service.GetCacheInstance().Set(ctx, cacheKey, cacheData, 0)

	// 清理临时表，但是不要每次都清理，以免io太多
	if deviceInfo.NetworkType == consts.ProductJointNetworkTypeLora {
		clearMapKey := dbConnGroup + consts.DistributeTableViWaveTemplate
		lastTime, ok := LastExecutedClearTime[clearMapKey]
		if !ok || time.Now().Unix()-lastTime.Unix() > int64(ClearSeconds) {
			LastExecutedClearTime[clearMapKey] = time.Now()
			past := time.Now().Add(-time.Duration(ClearSeconds) * time.Second)
			timeLayout := "2006-01-02 15:04:05"
			timeStr := past.Local().Format(timeLayout)
			_, err = db.Model(consts.DistributeTableViWaveTemplate).Ctx(ctx).Where(dao.IotViWaveDemoTemplate.Columns().CreatedAt+"<", timeStr).Delete()
			if err != nil {
				return err
			}
		}
	}

	return nil
}

func (s *sAeemqx) handleLevelData(ctx context.Context, message *pb.SensorMessage) (err error) {
	if message == nil {
		return
	}
	if message.GetLevelResult() == nil {
		return
	}
	levelData := message.GetLevelResult()
	sn := message.GetNodeId()
	// 可选值 strength  synthesis 或者 “1”和 “2”
	type1 := 0
	if strings.ToLower(levelData.GetType()) == "strength" || levelData.GetType() == "1" {
		type1 = 1
	} else if strings.ToLower(levelData.GetType()) == "synthesis" || levelData.GetType() == "2" {
		type1 = 2
	}
	if type1 == 0 {
		return
	}
	glog.Line().Info(ctx, sn, type1)
	return
}

func (s *sAeemqx) handleAeAstResult(ctx context.Context, message *pb.SensorMessage) (err error) {
	if message == nil {
		err = errors.New("invalid message")
		return
	}
	if message.GetAeAstResult() == nil {
		err = errors.New("invalid ast result")
		return
	}
	deviceNo := message.GetNodeId()
	deviceInfo, err := service.Device().GetByDeviceNo(ctx, deviceNo)
	if err != nil {
		return err
	}
	if deviceInfo.OpenStatus != consts.OpenStatusActive {
		err = errors.New("device is not open")
		return err
	}
	secMicro := timeUtil.GetJoinSecNsToMicroSec(uint64(message.GetAeAstResult().GetArrivingSec()), uint64(message.GetAeAstResult().GetArrivingNs()))
	_, err = dao.IotDeviceAstLog.Ctx(ctx).Insert(do.IotDeviceAstLog{
		DeviceNo:   deviceNo,
		DeptId:     deviceInfo.DeptId,
		CreatedAt:  gtime.Now(),
		Data:       message.GetAeAstResult().GetAstData(),
		Detail:     nil,
		Sampletime: secMicro,
	})
	if err != nil {
		return err
	}

	// 清理该设备10分钟以前的ast记录，但至少保留一条最新的
	s.cleanupOldAstRecords(ctx, deviceNo)

	return
}

// cleanupOldAstRecords 清理旧的AST记录，保留最新的一条
func (s *sAeemqx) cleanupOldAstRecords(ctx context.Context, deviceNo string) {
	// 检查是否需要清理（10分钟内只清理一次）
	astCleanupMutex.Lock()
	lastCleanupTime, exists := astCleanupCache[deviceNo]
	now := time.Now()
	// 如果不存在记录或者已经超过10分钟，则执行清理
	if !exists || now.Sub(lastCleanupTime) > 10*time.Minute {
		// 更新最后清理时间
		astCleanupCache[deviceNo] = now
		astCleanupMutex.Unlock()

		// 获取10分钟前的时间戳
		tenMinutesAgo := now.Add(-10 * time.Minute)

		// 先获取该设备最新的一条记录ID，确保不会被删除
		var latestRow *entity.IotDeviceAstLog
		err := dao.IotDeviceAstLog.Ctx(ctx).
			Where("device_no = ?", deviceNo).
			Order("sampletime DESC").
			Limit(1).Scan(&latestRow)

		if err == nil && latestRow.Id > 0 {
			// 删除10分钟前的记录，但保留最新的一条
			_, err = dao.IotDeviceAstLog.Ctx(ctx).
				Where("device_no = ? AND sampletime < ? AND id != ?", deviceNo, tenMinutesAgo.UnixNano()/1000, latestRow.Id).
				Delete()
		}
	} else {
		astCleanupMutex.Unlock()
	}
}
